package com.dxf.bigdata.D04_spark

import java.io.{BufferedInputStream, ObjectOutputStream, OutputStream}
import java.net.Socket

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.io.StdIn

/**
 *  RDD :  弹性分布式数据集
 *   类比 T01中的Task
 *
 *                                       Task (理解为 T01_SubTask)                   Executor
 *   driver(RDD(可以理解为T01_Task))   ----------------------------------------->
 *                                                                                   Executor
 *
 *
 *   RDD 最小的计算单元,代表一个弹性,不可变,可分区,;面元素可并行的计算集合
 *
 *   弹性:
 *      1 存储弹性 :内存和磁盘自动切换
 *      2 容错弹性 :数据丢失自动回复
 *      3 计算弹性 :计算出错重试机制
 *      4 分片弹性 :可根据需求重新分片
 *   分布式:
 *      数据存储在大数据集群不同节点上
 *   数据集:
 *      RDD封装了计算逻辑,并不保存数据
 *   数据抽象:
 *      RDD是一个抽象类,需要子类具体实现
 *   不可变:
 *      RDD封装了计算逻辑,是不可改变的,想改变只能生产新的RDD
 *   可分区并行计算
 *
 *   =============================
 *   RDD 5个属性
 *  Internally, each RDD is characterized by five main properties:
 *  *
 *  *  - A list of partitions
 *  *  - A function for computing each split
 *  *  - A list of dependencies on other RDDs
 *  *  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
 *  *  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
 *  *    an HDFS file)
 *
 *   1 分区列表
 *   2 分区计算函数
 *   3 RDD依赖关系 链表
 *   4 分区器  数据如何分区
 *   5 首选位置   数据发送给哪一个Executor--> 发送给哪一个节点进行计算效率最优(如 移动数据不如移动计算)   getPreferredLocations
 *
 *   =====================================================
 *
 *
 *   RDD 创建:
 *     1 从内存中创建RDD
 *
 */
object T02_RDD {
  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    //TODO 创建RDD
    //从内存创建RDD 将内存中集合的数据作为数据源
    val seq = Seq[Int](1, 2, 3, 4)
    // parallelize 并行
    val rdd: RDD[Int] = sc.parallelize(seq)

    val radd2: RDD[Int] = sc.makeRDD(seq)

    rdd.collect().foreach(println)
    //TODO 关闭环境
    sc.stop()


  }
}
